package main

import (
	"bufio"
	"fmt"
	"os"
	"strconv"

	"./pipelineMiddleWare"
)

// 本地归并排序

func main() {
	//ch1 := createPipeline("big.in", 10200, 4)

	ch1 := createNetworkPipeline("big.in", 10200, 4)

	pipelineMiddleWare.UseTime()

	writetofile(ch1, "big.out")

	showfile("big.out")

}

//网络版调用
func createNetworkPipeline(fileName string, fileSize int, chunkCount int) <-chan int {
	file, err := os.Create(fileName)
	if err != nil {
		panic(err)
	}
	defer file.Close() //延迟关闭文件

	mypipe := pipelineMiddleWare.RandomSource(fileSize / 8) //管道装的随机数

	writer := bufio.NewWriter(file)

	pipelineMiddleWare.WriterSlink(writer, mypipe)
	writer.Flush()

	chunkSize := fileSize / chunkCount //大小
	sortAddr := []string{}             // 分布式地址

	pipelineMiddleWare.Init()

	file, err = os.Open(fileName)
	if err != nil {
		panic(err)
	}
	defer file.Close() //延迟关闭文件

	for i := 0; i < chunkCount; i++ {
		file.Seek(int64(i*chunkSize), 0) //跳到文件指针
		source := pipelineMiddleWare.ReaderSource(bufio.NewReader(file), chunkSize)

		addr := ":" + strconv.Itoa(7000+i) //开辟地址

		pipelineMiddleWare.NetWordkWrite(addr, pipelineMiddleWare.InMemorySort(source))
		sortAddr = append(sortAddr, addr)
	}

	sortResults := []<-chan int{} // //排序结果，一个数组，每一个元素是个管道

	for _, addr := range sortAddr {
		sortResults = append(sortResults, pipelineMiddleWare.NetWordkRead(addr))
	}

	return pipelineMiddleWare.MergeN(sortResults...)
}

//多线程-调用中间件完成
// filename  文件名
//fileSize   文件大小
// chunkCount 分割块
func createPipeline(fileName string, fileSize int, chunkCount int) <-chan int {
	file, err := os.Create(fileName)
	if err != nil {
		panic(err)
	}
	defer file.Close() //延迟关闭文件

	mypipe := pipelineMiddleWare.RandomSource(fileSize / 8) //管道装的随机数

	writer := bufio.NewWriter(file)

	pipelineMiddleWare.WriterSlink(writer, mypipe)
	writer.Flush()

	chunkSize := fileSize / chunkCount // 数量

	sortResults := []<-chan int{} // //排序结果，一个数组，每一个元素是个管道

	pipelineMiddleWare.Init()

	file, err = os.Open(fileName)
	if err != nil {
		panic(err)
	}
	defer file.Close() //延迟关闭文件

	for i := 0; i < chunkCount; i++ {
		fmt.Println(i)

		file.Seek(int64(i*chunkSize), 0) //跳到文件指针
		source := pipelineMiddleWare.ReaderSource(bufio.NewReader(file), chunkSize)

		//for v := range source {
		//	fmt.Println(v)
		//}

		//break
		sortResults = append(sortResults, pipelineMiddleWare.InMemorySort(source))
	}

	return pipelineMiddleWare.MergeN(sortResults...)
}

//
func writetofile(in <-chan int, fileName string) {
	file, err := os.Create(fileName)
	if err != nil {
		panic(err)
	}
	defer file.Close()

	writer := bufio.NewWriter(file)
	defer writer.Flush() //刷新

	pipelineMiddleWare.WriterSlink(writer, in) //写入数据
}

func showfile(fileName string) {
	file, err := os.Open(fileName) //打开文件
	if err != nil {
		panic(err)
	}
	defer file.Close()

	p := pipelineMiddleWare.ReaderSource(bufio.NewReader(file), -1)

	counter := 0
	for v := range p {
		fmt.Println(v)
		counter++
		if counter > 1000 {
			break
		}
	}
}
